package rpc.serialize.kyro;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import rpc.exceprion.SerializeException;
import rpc.remote.dto.RpcRequest;
import rpc.remote.dto.RpcResponse;
import rpc.serialize.Serializer;


import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * @ClassName: KyroSerializer
 * @Description: 使用Kyro进行序列化,由于Kryo不是线程安全的，每个线程都有自己的input和output实例，所以要使用ThreadLocal存放。
 * @author: Kowk
 * @since:1.0.0
 */
public class KryoSerializer implements Serializer {

    private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() ->{
        Kryo kryo = new Kryo();
        kryo.register(RpcResponse.class);
        kryo.register(RpcRequest.class);
        kryo.setReferences(true);//默认为TRUE，是否关闭注册行为，如果关闭后可能存在存在序列化问题，推荐设置为TRUE
        kryo.setRegistrationRequired(false);//默认为FALSE，是否关闭循环引用，开启可以提高性能但一般不推荐开启
        return kryo;
            }
    );

    @Override
    public byte[] serialize(Object object)  {
        try(ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
            Output output = new Output(byteArrayOutputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            //Object->byte,将对象序列化为byte
            kryo.writeObject(output,object);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (IOException e) {
            throw new SerializeException("序列化失败");
        }
    }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> clazz) {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
             Input input = new Input(byteArrayInputStream)){
            Kryo kryo = kryoThreadLocal.get();
            //byte -> Object,将byte流反序列化为对象
            Object obj = kryo.readObject(input,clazz);
            kryoThreadLocal.remove();
            return clazz.cast(obj);
        } catch (IOException e) {
            throw new SerializeException("反序列化失败！");
        }
    }
}
